package com.rocketmq.example;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

public class E04_ClusterConsumer02 {


    /**
     * 多个消费者，同时使用*多个*消费者组，订阅同一个主题，消息会被每个消费组*分别*全部消费一遍
     * 
     */

    public static void main(String[] args) {
        consumer1();
        consumer2();

    }

    public static void consumer2() {
        // 通过push模式消费消息，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group-1");

        try {
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(ExampleMQConfig.TOPIC.getName(), "*");

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                // setConsumeMessageBatchMaxSize 就是指 List<MessageExt> msgs 的大小
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.printf("消费消息:%s", "group-1" + new String(msg.getBody()) + "\n");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void consumer1() {
        // 通过push模式消费消息，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group-2");

        try {
            // 指定NameServer的地址
            consumer.setNamesrvAddr(ExampleMQConfig.ENDPOINT.getName());

            // 订阅这个topic下的所有的消息
            consumer.subscribe(ExampleMQConfig.TOPIC.getName(), "*");

            // 注册一个消费的监听器，当有消息的时候，会回调这个监听器来消费消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                // setConsumeMessageBatchMaxSize 就是指 List<MessageExt> msgs 的大小
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                        ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        System.out.printf("消费消息:%s", "group-2" + new String(msg.getBody()) + "\n");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
